package com.xixi.canal.client.starter.loader;

import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.server.exception.CanalServerException;
import com.google.common.base.Function;
import com.google.common.collect.MigrateMap;
import com.xixi.canal.client.starter.config.CanalConfig;
import com.xixi.canal.client.starter.config.InstanceConfig;
import com.xixi.canal.client.starter.config.SpringContext;
import com.xixi.canal.client.support.YmlConfigLoader;
import com.xixi.canal.client.util.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.ConcurrentMap;

public class CanalClientStarter {
    protected static final Logger logger = LoggerFactory.getLogger(CanalClientStarter.class);

    private CanalConfig canalConfig;

    private ConcurrentMap<String, InstanceConfig> instancesConfigMap;

    private CanalServerWithEmbedded embeddedServer;

    private CanalClientLoader canalClientLoader;

    public CanalClientStarter(CanalConfig canalConfig) {
        this.canalConfig = canalConfig;
    }

    public synchronized void start() {

        logger.info("## start the canal server.");

        String destinations = canalConfig.getDestinations();

        Assert.notNull(destinations, "Instance cannot be empty.");

        String[] instances = destinations.split(",");

        Properties envProperties = SpringContext.getEnvProperties();


        instancesConfigMap = MigrateMap.makeComputingMap(new Function<String, InstanceConfig>() {
            @Override
            public InstanceConfig apply(String destination) {
                return YmlConfigLoader.loadYaml(envProperties, canalConfig.getConfDir(), destination, InstanceConfig.class);
            }
        });

        embeddedServer = CanalServerWithEmbedded.instance();

        embeddedServer.setCanalInstanceGenerator(new CanalInstanceGenerator() {
            @Override
            public CanalInstance generate(String destination) {

                InstanceConfig instanceConfig = instancesConfigMap.get(destination);

                if (null == instanceConfig) {
                    throw new CanalServerException("can't find destination:" + destination);
                }

                ClientCanalInstanceGenerator canalInstanceGenerator = new ClientCanalInstanceGenerator(canalConfig,
                        instanceConfig);

                return canalInstanceGenerator.generate(destination);
            }
        });

        embeddedServer.start();

        for (String instance : instances) {
            if (!embeddedServer.isStart(instance)) {
                embeddedServer.start(instance);
            }
        }
        logger.info("## the canal server is running now ......");

        logger.info("## start the canal client.");
        canalClientLoader = new CanalClientLoader(canalConfig, embeddedServer, instancesConfigMap);

        canalClientLoader.init();

        logger.info("## the canal client is running now ......");
    }

    public synchronized void stop() {
        logger.info("## stop the canal server.");

        if (null != canalClientLoader) {
            canalClientLoader.destroy();
            canalClientLoader = null;
        }

        if (null != instancesConfigMap) {
            instancesConfigMap.clear();
        }

        if (null != embeddedServer) {
            embeddedServer.stop();
        }
        logger.info("## the canal server is stopped ......");
    }

    public void refresh(String destination, String fileName) {

        if (instancesConfigMap.containsKey(destination)) {
            instancesConfigMap.remove(destination);
        }

        if (embeddedServer.isStart(destination)) {
            embeddedServer.stop(destination);
        }
        if (!embeddedServer.isStart(destination)) {
            embeddedServer.start(destination);
        }
    }
}
